package com.lagou.kafka.demo.serializer;

import com.lagou.kafka.demo.entity.User;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serializer;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.Map;

public class UserSerializer implements Serializer<User> {
    public void configure(Map<String, ?> configs, boolean isKey) {

    }

    public byte[] serialize(String s, User data) {
        try {
            // 如果数据是null，则返回null
            if(data == null) return null;
            Integer userId = data.getUserId();
            String username = data.getUsername();
            int length = 0;
            byte[] bytes = null;
            if(null != username){
                bytes = username.getBytes("utf-8");
            }
            length = bytes.length;
            ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + length);
            buffer.putInt(userId);
            buffer.putInt(length);
            buffer.put(bytes);
            return buffer.array();
        }catch (UnsupportedEncodingException e){
            throw new SerializationException("序列化异常");
        }

    }


    public void close() {

    }
}
